/**
 * Copyright 1999-2015 dangdang.com.
 * <p>
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * <p>
 * http://www.apache.org/licenses/LICENSE-2.0
 * <p>
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 * </p>
 */

package spring.cloud.tasks.api.service.impl;

import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
import org.apache.commons.lang3.StringUtils;
import org.springframework.util.CollectionUtils;
import spring.cloud.tasks.api.domain.*;
import spring.cloud.tasks.api.service.ExecutorService;
import spring.cloud.tasks.api.service.TaskService;
import spring.cloud.tasks.api.service.ZooKeeperService;
import spring.cloud.tasks.api.utils.ExecutorNodePath;
import spring.cloud.tasks.api.utils.TaskNodePath;
import spring.cloud.tasks.api.zookeeper.ZooKeeper;
import spring.cloud.tasks.api.zookeeper.ZooKeeperRepository;

import javax.annotation.Resource;
import java.util.List;
import java.util.Set;

/**
 * Default implementation of ExecutorService.
 */
public class ExecutorServiceImpl implements ExecutorService {

    private static final int DEFAULT_MAX_SECONDS_FORCE_KILL_EXECUTOR = 300;
    private static final int SMALLEST_VERSION_SUPPORTED_DUMP = 3;
    private static final Set<String> SUPPORT_DUMP_VERSION_WHITE_LIST = Sets.newHashSet("saturn-dev", "master-SNAPSHOT");

    @Resource
    private ZooKeeperRepository zooKeeperRepository;

    @Resource
    private TaskService taskService;

    @Resource
    private ZooKeeperService zooKeeperService;


    @Override
    public List<ExecutorInfo> getExecutorInfoList(String namespace) throws RuntimeException {
        return getExecutorInfoList(namespace, null);
    }

    @Override
    public List<ExecutorInfo> getExecutorInfoList(String namespace, TaskExecutorStatus12 executorStatus)
            throws RuntimeException {
        ZooKeeper zooKeeper = getZooKeeper(namespace);

        List<String> executorIdList = zooKeeper.getChildren(ExecutorNodePath.getExecutorNodePath());
        if (executorIdList == null || executorIdList.isEmpty()) {
            return Lists.newArrayList();
        }

        List<ExecutorInfo> executorInfoList = Lists.newArrayList();
        for (String executorId : executorIdList) {
            ExecutorInfo executorInfo = getExecutorInfo(executorId, zooKeeper);
            if (executorStatus == null || executorInfo.getStatus() == executorStatus) {
                executorInfoList.add(executorInfo);
            }
        }

        return executorInfoList;
    }

    @Override
    public ExecutorInfo getExecutorInfo(String namespace, String executorId) throws RuntimeException {
        ZooKeeper zooKeeper = getZooKeeper(namespace);

        if (!zooKeeper.checkExists(ExecutorNodePath.getExecutorNodePath(executorId))) {
            return null;
        }

        return getExecutorInfo(executorId, zooKeeper);
    }

    private ExecutorInfo getExecutorInfo(String executorId, ZooKeeper zooKeeper) {
        ExecutorInfo executorInfo = new ExecutorInfo(executorId);
        String ip = zooKeeper.getData(ExecutorNodePath.getExecutorIpNodePath(executorId));
        executorInfo.setServerIp(ip);
        if (StringUtils.isNotBlank(ip)) {
            executorInfo.setStatus(TaskExecutorStatus12.ONLINE);
        } else {
            executorInfo.setStatus(TaskExecutorStatus12.OFFLINE);
        }

        String restartNodePath = ExecutorNodePath.getExecutorRestartNodePath(executorId);
        long restartTriggerTime = zooKeeper.getCtime(restartNodePath);
        long now = System.currentTimeMillis();

        long maxRestartInv = 1000;

        // 如果restart结点存在，restart结点存在的时间<300s，executor状态列显示RESTARTING；
        executorInfo.setRestarting(0 != restartTriggerTime && now - restartTriggerTime < maxRestartInv);

        // 是否已被摘流量
        executorInfo.setNoTraffic(zooKeeper.checkExists(ExecutorNodePath.getExecutorNoTrafficNodePath(executorId)));
        // lastBeginTime
        String lastBeginTime = zooKeeper.getData(ExecutorNodePath.getExecutorNodePath(executorInfo.getExecutorName(), "lastBeginTime"));
        executorInfo.setLastBeginTime(lastBeginTime);

        String containerGroupName = zooKeeper.getData(ExecutorNodePath.getExecutorTaskNodePath(executorId));
        if (StringUtils.isNotBlank(containerGroupName)) {
            // 是否容器
            executorInfo.setContainer(true);
            // 容器组
            executorInfo.setContainerGroupName(containerGroupName);
        }
        return executorInfo;
    }

    @Override
    public ExecutorAllocatInfo getExecutorAllocation(String nameSpace, String executorId) throws RuntimeException {
        ZooKeeper zooKeeper = getZooKeeper(nameSpace);

        List<TaskConfiguration> taskConfigurationList = taskService.getTaskConfigurationList(nameSpace);

        ExecutorAllocatInfo executorAllocateInfo = new ExecutorAllocatInfo(executorId);

        for (TaskConfiguration taskConfiguration : taskConfigurationList) {
            String taskId = taskConfiguration.getTaskId();
            String serverNodePath = TaskNodePath.getExecutorsNodePath(taskId);
            if (!zooKeeper.checkExists(serverNodePath)) {
                continue;
            }

            String sharding = zooKeeper.getData(TaskNodePath.getExecutorsNodePath(taskId, executorId, "sharding"));
            if (StringUtils.isNotBlank(sharding)) {
                // 作业状态为STOPPED的即使有残留分片也不显示该分片
                if (TaskStatus.STOPPED.equals(taskService.getTaskStatus(nameSpace, taskId))) {
                    continue;
                }
                // concat executorSharding
                executorAllocateInfo.getAllocationMap().put(taskId, sharding);
                // calculate totalLoad
                String loadLevelNode = zooKeeper.getData(TaskNodePath.getConfigNodePath(taskId, "loadLevel"));
                Integer loadLevel = 1;
                if (StringUtils.isNotBlank(loadLevelNode)) {
                    loadLevel = Integer.valueOf(loadLevelNode);
                }

                int shardingItemNum = sharding.split(",").length;
                int taskLoad = shardingItemNum * loadLevel;
                int totalLoad = executorAllocateInfo.getTotalLoadLevel();
                executorAllocateInfo.setTotalLoadLevel(totalLoad + taskLoad);
            }
        }

        return executorAllocateInfo;
    }

    @Override
    public ExecutorRunningInfo getExecutorRunningInfo(String nameSpace, String executorId)
            throws RuntimeException {
        ZooKeeper zooKeeper = getZooKeeper(nameSpace);
        List<TaskConfiguration> taskConfigurationList = taskService.getTaskConfigurationList(nameSpace);
        ExecutorRunningInfo executorRunningInfo = new ExecutorRunningInfo();
        executorRunningInfo.setExecutorId(executorId);
        for (TaskConfiguration taskConfiguration : taskConfigurationList) {


            String taskId = taskConfiguration.getTaskId();
            String serverNodePath = TaskNodePath.getExecutorsNodePath(taskId);
            if (!zooKeeper.checkExists(serverNodePath)) {
                continue;
            }

            String sharding = zooKeeper.getData(TaskNodePath.getExecutorsNodePath(taskId, executorId, "sharding"));
            Set<String> shardingItems = getShardingItems(sharding);
            boolean needToCheckFailover = needToCheckFailover(taskConfiguration);
            if (needToCheckFailover) {
                obtainServerRunningInfoWhileNeedToCheckFailover(executorId, zooKeeper, taskId, shardingItems, executorRunningInfo);
            } else if (!shardingItems.isEmpty()) {
                obtainServerRunningInfoWhileNoNeedToCheckFailover(zooKeeper, taskConfiguration, shardingItems, executorRunningInfo);
            }

        }

        return executorRunningInfo;
    }


    /**
     * 不需要检查潜在运行的作业，因为这类作业不可能failover;
     * 遍历所有作业的所有分片，如果存在failover节点，则要看看有没有running节点，如果有running节点则要看看是否自己正在执行failover，如果是加入到runningInfo中，否则continue；
     */
    private void obtainServerRunningInfoWhileNeedToCheckFailover(String executorId,
                                                                 ZooKeeper zooKeeper,
                                                                 String taskId,
                                                                 Set<String> shardingItems,
                                                                 ExecutorRunningInfo executorRunningInfo) {
        List<String> executionItems = zooKeeper.getChildren(TaskNodePath.getExecutionNodePath(taskId));
        if (CollectionUtils.isEmpty(executionItems)) {
            return;
        }

        List<String> runningItems = Lists.newArrayList();
        for (String item : executionItems) {
            boolean isItemRunning = zooKeeper.checkExists(TaskNodePath.getExecutionNodePath(taskId, item, "running"));
            if (!isItemRunning) {
                continue;
            }
            // 属于已经分配的分片
            if (shardingItems != null && shardingItems.contains(item)) {
                runningItems.add(item);
                continue;
            }
            // 不属于自己分配的分片，则看看是否有failover
            String failoverValue = zooKeeper.getData(TaskNodePath.getExecutionNodePath(taskId, item, "failover"));
            if (StringUtils.isNotBlank(failoverValue) && failoverValue.equals(executorId)) {
                runningItems.add(item);
            }
        }

        if (!runningItems.isEmpty()) {
            executorRunningInfo.getRunningJobItems().put(taskId, StringUtils.join(runningItems, ','));
        }
    }

    private void obtainServerRunningInfoWhileNoNeedToCheckFailover(ZooKeeper zooKeeper, TaskConfiguration taskConfiguration, Set<String> shardingItems, ExecutorRunningInfo serverRunningInfo) {
        // 对于不上报运行状态的作业，所有分配的分片均作为潜在运行中的作业


        List<String> runningItems = Lists.newArrayList();
        for (String item : shardingItems) {
            if (zooKeeper.checkExists(TaskNodePath.getExecutionNodePath(taskConfiguration.getTaskId(), item, "running"))) {
                runningItems.add(item);
            }
        }

        if (!runningItems.isEmpty()) {
            serverRunningInfo.getRunningJobItems().put(taskConfiguration.getTaskId(), StringUtils.join(runningItems, ','));
        }
    }

    private Set<String> getShardingItems(String sharding) {
        if (StringUtils.isBlank(sharding)) {
            return Sets.newHashSet();
        }

        String[] items = sharding.split(",");
        Set<String> result = Sets.newTreeSet();
        for (String item : items) {
            if (StringUtils.isBlank(item)) {
                continue;
            }
            result.add(item.trim());
        }
        return result;
    }

    private boolean needToCheckFailover(TaskConfiguration taskConfiguration) {
        return taskConfiguration.getFailover();
    }

    @Override
    public void extractTraffic(String nameSpace, String executorId) throws RuntimeException {
        ZooKeeper zooKeeper = getZooKeeper(nameSpace);
        validateIfExecutorNameExisted(executorId, zooKeeper);
        zooKeeper.create(ExecutorNodePath.getExecutorNoTrafficNodePath(executorId));
    }

    @Override
    public void recoverTraffic(String nameSpace, String executorId) throws RuntimeException {
        ZooKeeper zooKeeper = getZooKeeper(nameSpace);
        validateIfExecutorNameExisted(executorId, zooKeeper);
        zooKeeper.deleteRecursive(ExecutorNodePath.getExecutorNoTrafficNodePath(executorId));
    }

    @Override
    public void removeExecutor(String nameSpace, String executorId) throws RuntimeException {
        ZooKeeper zooKeeper = getZooKeeper(nameSpace);
        zooKeeper.deleteRecursive(ExecutorNodePath.getExecutorNodePath(executorId));
        List<String> taskIdList = taskService.getTaskIdFromZooKeeper(nameSpace);
        if (CollectionUtils.isEmpty(taskIdList)) {
            return;
        }
        for (String taskId : taskIdList) {
            String executorNode = TaskNodePath.getExecutorsNodePath(taskId, executorId);
            zooKeeper.deleteRecursive(executorNode);
        }
    }

    @Override
    public void shardAll(String nameSpace) throws RuntimeException {
        ZooKeeper zooKeeper = getZooKeeper(nameSpace);
        String shardAllAtOnceNodePath = ExecutorNodePath.getExecutorShardingNodePath("shardAllAtOnce");
        zooKeeper.deleteRecursive(shardAllAtOnceNodePath);
        zooKeeper.create(shardAllAtOnceNodePath);
    }


    @Override
    public void restart(String nameSpace, String executorId) throws RuntimeException {
        ZooKeeper zooKeeper = getZooKeeper(nameSpace);
        String restartNodePath = ExecutorNodePath.getExecutorRestartNodePath(executorId);
        zooKeeper.delete(restartNodePath);
        zooKeeper.create(restartNodePath);
    }

    private void validateIfExecutorNameExisted(String executorName, ZooKeeper zooKeeper) throws RuntimeException {
        if (!zooKeeper.checkExists(ExecutorNodePath.getExecutorNodePath(executorName))) {
            throw new RuntimeException("The executorName(" + executorName + ") is not existed.");
        }
    }

    protected ZooKeeper getZooKeeper(String namespace) throws RuntimeException {
        return zooKeeperService.getZooKeeper(namespace);
    }
}
